/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.zookeeper;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.LinkedBlockingQueue;

import javax.security.auth.login.LoginException;
import javax.security.sasl.SaslException;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.Record;
import org.apache.zookeeper.AsyncCallback.ACLCallback;
import org.apache.zookeeper.AsyncCallback.Children2Callback;
import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.MultiCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.OpResult.ErrorResult;
import org.apache.zookeeper.Watcher.Event;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.ZooKeeper.WatchRegistration;
import org.apache.zookeeper.client.HostProvider;
import org.apache.zookeeper.client.ZooKeeperSaslClient;
import org.apache.zookeeper.common.Time;
import org.apache.zookeeper.proto.AuthPacket;
import org.apache.zookeeper.proto.ConnectRequest;
import org.apache.zookeeper.proto.CreateResponse;
import org.apache.zookeeper.proto.ExistsResponse;
import org.apache.zookeeper.proto.GetACLResponse;
import org.apache.zookeeper.proto.GetChildren2Response;
import org.apache.zookeeper.proto.GetChildrenResponse;
import org.apache.zookeeper.proto.GetDataResponse;
import org.apache.zookeeper.proto.GetSASLRequest;
import org.apache.zookeeper.proto.ReplyHeader;
import org.apache.zookeeper.proto.RequestHeader;
import org.apache.zookeeper.proto.SetACLResponse;
import org.apache.zookeeper.proto.SetDataResponse;
import org.apache.zookeeper.proto.SetWatches;
import org.apache.zookeeper.proto.WatcherEvent;
import org.apache.zookeeper.server.ByteBufferInputStream;
import org.apache.zookeeper.server.ZooKeeperThread;
import org.apache.zookeeper.server.ZooTrace;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class manages the socket i/o for the client. ClientCnxn maintains a list
 * of available servers to connect to and "transparently" switches servers it is
 * connected to as needed.
 *
 * 管理与server端的链接。
 * 这个类维护了一个可用的服务端链接，在需要的时候，进行切换服务端链接。
 *
 */
public class ClientCnxn {
    /**
     * @see BinaryInputArchive#maxBuffer
     */
    public static final int packetLen = Integer.getInteger("jute.maxbuffer",4096 * 1024);
    private static final Logger LOG = LoggerFactory.getLogger(ClientCnxn.class);
    private static final String ZK_SASL_CLIENT_USERNAME = "zookeeper.sasl.client.username";
    /* ZOOKEEPER-706: If a session has a large number of watches set then
     * attempting to re-establish those watches after a connection loss may
     * fail due to the SetWatches request exceeding the server's configured
     * jute.maxBuffer value. To avoid this we instead split the watch
     * re-establishement across multiple SetWatches calls. This constant
     * controls the size of each call. It is set to 128kB to be conservative
     * with respect to the server's 1MB default for jute.maxBuffer.
     */
    private static final int SET_WATCHES_MAX_LENGTH = 128 * 1024;
    /**
     * 此变量控制‘自动watch重置’是否开启。
     * 当session重现连接时，客户端将自动重置watches，
     * 设置环境变量 zookeeper.disableAutoWatchReset = true，可以关闭这个行为。
     *
     * This controls whether[是否] automatic watch resetting is enabled.
     * Clients automatically reset watches during session reconnect, this
     * option allows the client to turn off this behavior by setting the environment variable
     * "zookeeper.disableAutoWatchReset" to "true"
     * */
    private static boolean disableAutoWatchReset;

    static {
        disableAutoWatchReset = Boolean.getBoolean("zookeeper.disableAutoWatchReset");
        if (LOG.isDebugEnabled()) {
            LOG.debug("zookeeper.disableAutoWatchReset is " + disableAutoWatchReset);
        }
    }

    final String chrootPath;

    final SendThread sendThread;

    final EventThread eventThread;

    private final CopyOnWriteArraySet<AuthData> authInfo = new CopyOnWriteArraySet<AuthData>();
    /**
     * 需要被发送的数据包[packet]
     */
    private final LinkedList<Packet> outgoingQueue = new LinkedList<Packet>();

    /**
     * 这些数据包已经发送并正在等待响应。
     */
    private final LinkedList<Packet> pendingQueue = new LinkedList<Packet>();

    /**
     * 客户端请求的session过期时间[the timeout request by the client]
     *
     * @see ClientCnxn#negotiatedSessionTimeout
     */
    private final int sessionTimeout;

    /**
     * The timeout in ms the client negotiated with the server.
     * This is the "real" timeout, not the timeout request by the client (which may have
     * been increased/decreased by the server which applies bounds to this value.
     * 服务端协商后的session过期时间。
     * 这是‘真正的’过期时间，不是客户端请求的过期时间。在sessionTimeout的基础上，可能增加也可能减小。
     * 具体范围在：   2*tickTime <= negotiatedSessionTimeout <= 20*tickTime
     */
    private volatile int negotiatedSessionTimeout;

    private final ZooKeeper zooKeeper;

    /**
     * 客户端watcher管理器
     *
     * 如果client需要监听一个path，则只需要告知server一个boolean类型的标识就行了。
     * 当数据有变化的时候，server会给client发送一个通知，但是具体数据变化之后，要执行什么动作，也就是watcher，
     * 是在client保存的。
     */
    private final ClientWatchManager watcher;
    /**
     * A set of ZooKeeper hosts this client could connect to.
     * 客户端将会连接的服务端域名集合
     */
    private final HostProvider hostProvider;
    public ZooKeeperSaslClient zooKeeperSaslClient;
    /**
     * Is set to true when a connection to a r/w server is established for the
     * first time; never changed afterwards.
     * <p>
     * Is used to handle situations when client without sessionId connects to a
     * read-only server. Such client receives "fake" sessionId from read-only
     * server, but this sessionId is invalid for other servers. So when such
     * client finds a r/w server, it sends 0 instead of fake sessionId during
     * connection handshake and establishes new, valid session.
     * <p>
     * If this field is false (which implies we haven't seen r/w server before)
     * then non-zero sessionId is fake, otherwise it is valid.
     *
     * 当用户在初始化时，提供了会话ID，就认为在初始化之前已经和服务端进行过read/write请求了
     */
    volatile boolean seenRwServerBefore = false;
    /**
     * 和服务端建立链接的超时时间
     * connectTimeout = sessionTimeout / hostProvider.size(); 或者
     * connectTimeout = negotiatedSessionTimeout / hostProvider.size();
     */
    private int connectTimeout;
    /**
     * 读操作超时时间
     * readTimeout = sessionTimeout * 2 / 3;   或者
     * readTimeout = negotiatedSessionTimeout * 2 / 3;
     */
    private int readTimeout;

    private long sessionId; // sessionId。
    private byte sessionPasswd[] = new byte[16]; //session密码
    /**
     * 如果为true：说明当前客户端与服务端的链接是read-only模式。
     * 在session创建握手时，除了其他字段，这个字段的值也将会被发送。
     * If true, the connection is allowed to go to r-o mode. This field's value
     * is sent, besides other data, during session creation handshake.
     * 如果服务端集群出现了分区，那么它（指分区后的服务端）将只接收只读客户端的请求。
     * If the server on the other side of the wire is partitioned it'll accept read-only clients only.
     */
    private boolean readOnly;
    /**
     * close方法被调用时，将会设置为true。
     */
    private volatile boolean closing = false;
    /**
     * 死亡事件 标识位
     */
    private Object eventOfDeath = new Object();
    private volatile long lastZxid;
    /**
     *
     * 客户端请求的事务id
     * 1、xid是自增的。server端就是根据这个xid来进行顺序执行事务。【ZooKeeper顺序一致性】   xid自增特性 详见：{@link ClientCnxn#getXid }
     * 2、有几个特殊的xid:   详见：{@link SendThread#readResponse }
     *   -1 means notification
     *   -2 is the xid for pings
     *   -4 is the xid for AuthPacket
     *   -8 表示：{@link ZooDefs.OpCode#setWatches}请求      详见：{@link SendThread#primeConnection }
     *
     */
    private int xid = 1;
    /**
     * 客户端状态；默认是 未连接
     */
    private volatile States state = States.NOT_CONNECTED;

    public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
                      ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, boolean canBeReadOnly) throws IOException {
        this(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher, clientCnxnSocket, 0, new byte[16], canBeReadOnly);
    }

    /**
     * Creates a connection object. The actual network connect doesn't get
     * established until needed. The start() instance method must be called
     * subsequent to construction.
     *
     * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
     * @param hostProvider the list of ZooKeeper servers to connect to
     * @param sessionTimeout the timeout for connections.
     * @param zooKeeper the zookeeper object that this connection is related to.
     * @param watcher watcher for this connection
     * @param clientCnxnSocket the socket implementation used (e.g. NIO/Netty)
     * @param sessionId session id if re-establishing session
     * @param sessionPasswd session passwd if re-establishing session
     * @param canBeReadOnly whether the connection is allowed to go to read-only mode in case of partitioning
     * @throws IOException
     */
    public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
                      ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket,
                      long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
        this.zooKeeper = zooKeeper;
        this.watcher = watcher;
        this.sessionId = sessionId;
        this.sessionPasswd = sessionPasswd;
        this.sessionTimeout = sessionTimeout;
        this.hostProvider = hostProvider;
        this.chrootPath = chrootPath;

        connectTimeout = sessionTimeout / hostProvider.size();
        readTimeout = sessionTimeout * 2 / 3;
        readOnly = canBeReadOnly;

        sendThread = new SendThread(clientCnxnSocket);
        eventThread = new EventThread();
    }

    /**
     * tests use this to check on reset of watches
     * @return if the auto reset of watches are disabled
     */
    public static boolean getDisableAutoResetWatch() {
        return disableAutoWatchReset;
    }

    /**
     * tests use this to set the auto reset
     * @param b the value to set disable watches to
     */
    public static void setDisableAutoResetWatch(boolean b) {
        disableAutoWatchReset = b;
    }

    /**
     * Guard against creating "-EventThread-EventThread-EventThread-..." thread
     * names when ZooKeeper object is being created from within a watcher.
     * See ZOOKEEPER-795 for details.
     */
    private static String makeThreadName(String suffix) {
        String name = Thread.currentThread().getName().replaceAll("-EventThread", "");
        return name + suffix;
    }

    public long getSessionId() {
        return sessionId;
    }

    public byte[] getSessionPasswd() {
        return sessionPasswd;
    }

    public int getSessionTimeout() {
        return negotiatedSessionTimeout;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();

        SocketAddress local = sendThread.getClientCnxnSocket().getLocalSocketAddress();
        SocketAddress remote = sendThread.getClientCnxnSocket().getRemoteSocketAddress();
        sb.append("sessionid:0x").append(Long.toHexString(getSessionId())).append(" local:").append(local).append(" remoteserver:").append(remote).append(" lastZxid:").append(lastZxid).append(" xid:").append(xid).append(" sent:").append(sendThread.getClientCnxnSocket().getSentCount()).append(" recv:").append(sendThread.getClientCnxnSocket().getRecvCount()).append(" queuedpkts:").append(outgoingQueue.size()).append(" pendingresp:").append(pendingQueue.size()).append(" queuedevents:").append(eventThread.waitingEvents.size());

        return sb.toString();
    }

    public void start() {
        sendThread.start();
        eventThread.start();
    }

    private void finishPacket(Packet p) {
        if (p.watchRegistration != null) {
            p.watchRegistration.register(p.replyHeader.getErr());
        }

        if (p.cb == null) {
            // 如果不存在异步callback方法：说明是同步阻塞请求。
            synchronized (p) {
                p.finished = true;
                // 唤醒在此请求上的等待
                p.notifyAll();
            }
        } else {
            p.finished = true;
            // 存在异步callback方法：说明是异步请求。
            eventThread.queuePacket(p);
        }
    }

    /**
     * 处理 客户端已断开，或者正在关闭状态下的packet
     *
     * @param conLossPacket
     */
    private void conLossPacket(Packet conLossPacket) {
        if (conLossPacket.replyHeader == null) {
            return;
        }
        switch (state) {
            case AUTH_FAILED:
                conLossPacket.replyHeader.setErr(KeeperException.Code.AUTHFAILED.intValue());
                break;
            case CLOSED:
                conLossPacket.replyHeader.setErr(KeeperException.Code.SESSIONEXPIRED.intValue());
                break;
            default:
                conLossPacket.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
        }
        // 完成packet
        finishPacket(conLossPacket);
    }

    public long getLastZxid() {
        return lastZxid;
    }

    /**
     * Shutdown the send/event threads. This method should not be called
     * directly - rather it should be called as part of close operation. This
     * method is primarily here to allow the tests to verify disconnection
     * behavior.
     */
    public void disconnect() {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Disconnecting client for session: 0x" + Long.toHexString(getSessionId()));
        }

        sendThread.close();
        eventThread.queueEventOfDeath();
    }

    /**
     * Close the connection, which includes; send session disconnect to the
     * server, shutdown the send/event threads.
     *
     * @throws IOException
     */
    public void close() throws IOException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Closing client for session: 0x" + Long.toHexString(getSessionId()));
        }

        try {
            RequestHeader h = new RequestHeader();
            h.setType(ZooDefs.OpCode.closeSession);

            submitRequest(h, null, null, null);
        } catch (InterruptedException e) {
            // ignore, close the send/event threads
        } finally {
            disconnect();
        }
    }

    /*
     * getXid() is called externally by ClientCnxnNIO::doIO() when packets are sent from the outgoingQueue to
     * the server. Thus, getXid() must be public.
     */
    synchronized public int getXid() {
        return xid++;
    }

    /**
     * 同步提交请求
     *
     * 注意：这里的 submitRequest() 方法和 {@link org.apache.zookeeper.server.quorum.ObserverZooKeeperServer#commitRequest(org.apache.zookeeper.server.Request) } 方法，意义不用。
     * 这里的submit值的是客户端网服务端提交请求。
     *
     * @param h
     * @param request
     * @param response
     * @param watchRegistration
     * @return
     * @throws InterruptedException
     */
    public ReplyHeader submitRequest(RequestHeader h, Record request,
                                     Record response, WatchRegistration watchRegistration) throws InterruptedException {
        // 创建响应头
        ReplyHeader r = new ReplyHeader();
        // 把请求放到队列中。
        Packet packet = queuePacket(h, r, request, response, null, null, null, null, watchRegistration);
        synchronized (packet) {
            // 如果数据包未处理完成，则一直等待。
            while (!packet.finished) {
                packet.wait();
            }
        }
        return r;
    }

    public void enableWrite() {
        sendThread.getClientCnxnSocket().enableWrite();
    }

    public void sendPacket(Record request, Record response, AsyncCallback cb, int opCode) throws IOException {
        // 生成xid，因为packet数据包马上就会被发送，通过调用下面的 sendThread.sendPacket(p)
        int xid = getXid();
        RequestHeader h = new RequestHeader();
        h.setXid(xid);
        h.setType(opCode);

        ReplyHeader r = new ReplyHeader();
        r.setXid(xid);

        Packet p = new Packet(h, r, request, response, null, false);
        p.cb = cb;

        // io发送线程，发送数据包
        sendThread.sendPacket(p);
    }

    Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
                       Record response, AsyncCallback cb, String clientPath,
                       String serverPath, Object ctx, WatchRegistration watchRegistration) {
        Packet packet = null;

        // 注意：我们还没为当前packet生成 requestHeader.xid。
        // 在ClientCnxnSocket子类的doIO()方法中：requestHeader.xid 将会在packet真正被发送时生成。【全局搜索生成'p.requestHeader.xid生成'可找到响应的代码】
        synchronized (outgoingQueue) {
            packet = new Packet(h, r, request, response, watchRegistration);
            packet.cb = cb;
            packet.ctx = ctx;
            packet.clientPath = clientPath;
            packet.serverPath = serverPath;
            // 如果提交完请求，客户端断开或即将关闭，则调用conLossPacket(packet)去处理socket连接丢失的packet
            if (!state.isAlive() || closing) {
                conLossPacket(packet);
            } else {
                // 如果客户端请求关闭session，则标记 closing
                if (h.getType() == OpCode.closeSession) {
                    closing = true;
                }
                // 入队
                outgoingQueue.add(packet);
            }
        }
        sendThread.getClientCnxnSocket().wakeupCnxn();
        return packet;
    }

    /**
     * 给当前客户端添加权限信息。acl相关知识点。
     * @param scheme
     * @param auth
     */
    public void addAuthInfo(String scheme, byte auth[]) {
        if (!state.isAlive()) {
            return;
        }
        authInfo.add(new AuthData(scheme, auth));
        RequestHeader h = new RequestHeader(-4, OpCode.auth);
        queuePacket(h, null, new AuthPacket(0, scheme, auth), null, null, null, null, null, null);
    }

    States getState() {
        return state;
    }

    static class AuthData {
        String scheme;
        byte data[];

        AuthData(String scheme, byte data[]) {
            this.scheme = scheme;
            this.data = data;
        }
    }

    /**
     * This class allows us to pass the headers and the relevant records around.
     */
    static class Packet {
        public boolean readOnly;
        RequestHeader requestHeader;
        ReplyHeader replyHeader;
        Record request;
        Record response;
        /**
         * packet实例数据的二级制格式
         * 详见：{@link Packet#createBB }
         */
        ByteBuffer bb;
        /** Client's view of the path (may differ due to chroot) **/
        String clientPath;
        /** Servers's view of the path (may differ due to chroot) **/
        String serverPath;
        boolean finished;
        AsyncCallback cb;
        Object ctx;
        WatchRegistration watchRegistration;

        Packet(RequestHeader requestHeader, ReplyHeader replyHeader, Record request, Record response, WatchRegistration watchRegistration) {
            this(requestHeader, replyHeader, request, response,watchRegistration, false);
        }

        Packet(RequestHeader requestHeader, ReplyHeader replyHeader, Record request, Record response, WatchRegistration watchRegistration, boolean readOnly) {
            this.requestHeader = requestHeader;
            this.replyHeader = replyHeader;
            this.request = request;
            this.response = response;
            this.readOnly = readOnly;
            this.watchRegistration = watchRegistration;
        }

        /**
         * 把数据包的内容写到byteBuffer中，即bb实例变量中
         */
        public void createBB() {
            try {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
                // 前四个字节是int类型，表示数据长度。这里先写成-1作为占位符，稍后将会填充真实数据长度。
                boa.writeInt(-1, "len");
                if (requestHeader != null) {
                    requestHeader.serialize(boa, "header");
                }
                if (request instanceof ConnectRequest) {
                    request.serialize(boa, "connect");
                    // append "am-I-allowed-to-be-readonly" flag
                    boa.writeBool(readOnly, "readOnly");
                } else if (request != null) {
                    request.serialize(boa, "request");
                }
                baos.close();
                // 使用byte[]包装声场的byteBuffer，position=0，所以下面一行可以从头直接写数据
                this.bb = ByteBuffer.wrap(baos.toByteArray());
                // 这里填充数据长度（减去4，是因为‘数据长度’占四字节）
                this.bb.putInt(this.bb.capacity() - 4);
                this.bb.rewind();
            } catch (IOException e) {
                LOG.warn("Ignoring unexpected exception", e);
            }
        }

        @Override
        public String toString() {
            StringBuilder sb = new StringBuilder();

            sb.append("clientPath:" + clientPath);
            sb.append(" serverPath:" + serverPath);
            sb.append(" finished:" + finished);

            sb.append(" header:: " + requestHeader);
            sb.append(" replyHeader:: " + replyHeader);
            sb.append(" request:: " + request);
            sb.append(" response:: " + response);

            // jute toString is horrible, remove unnecessary newlines
            return sb.toString().replaceAll("\r*\n+", " ");
        }
    }

    private static class WatcherSetEventPair {
        private final Set<Watcher> watchers;
        private final WatchedEvent event;

        public WatcherSetEventPair(Set<Watcher> watchers, WatchedEvent event) {
            this.watchers = watchers;
            this.event = event;
        }
    }

    static class EndOfStreamException extends IOException {
        private static final long serialVersionUID = -5438877188796231422L;

        public EndOfStreamException(String msg) {
            super(msg);
        }

        @Override
        public String toString() {
            return "EndOfStreamException: " + getMessage();
        }
    }

    private static class SessionTimeoutException extends IOException {
        private static final long serialVersionUID = 824482094072071178L;

        public SessionTimeoutException(String msg) {
            super(msg);
        }
    }

    private static class SessionExpiredException extends IOException {
        private static final long serialVersionUID = -1388816932076193249L;

        public SessionExpiredException(String msg) {
            super(msg);
        }
    }

    private static class RWServerFoundException extends IOException {
        private static final long serialVersionUID = 90431199887158758L;

        public RWServerFoundException(String msg) {
            super(msg);
        }
    }

    class EventThread extends ZooKeeperThread {

        /**
         * 等待被触发的事件
         */
        private final LinkedBlockingQueue<Object> waitingEvents = new LinkedBlockingQueue<>();

        /** This is really the queued session state until the event
         * thread actually processes the event and hands it to the watcher.
         * But for all intents and purposes this is the state.
         */
        private volatile KeeperState sessionState = KeeperState.Disconnected;

        private volatile boolean wasKilled = false;
        /**
         * eventThread是否仍在运行 标识
         */
        private volatile boolean isRunning = false;

        EventThread() {
            super(makeThreadName("-EventThread"));
            setDaemon(true);
        }

        /**
         * 监听事件 入队列
         * @param event
         */
        public void queueEvent(WatchedEvent event) {
            // 判断客户端session的状态。
            if (event.getType() == EventType.None && sessionState == event.getState()) {
                return;
            }
            sessionState = event.getState();

            // 从watcher管理器中，取出对应的 watcher
            Set<Watcher> watchers = watcher.materialize(event.getState(), event.getType(), event.getPath());
            // watcheres  和  event  的对应关系
            WatcherSetEventPair pair = new WatcherSetEventPair(watchers, event);

            // 入队
            waitingEvents.add(pair);
        }

        /**
         * Packet 入waitingEvents队列
         *
         * @param packet
         */
        public void queuePacket(Packet packet) {
            // 如果客户端将要关闭，则使用当前线程立即去处理事件;否则 就把packet入队，使用‘EventThread’线程去处理。
            if (wasKilled) {
                synchronized (waitingEvents) { // todo-lxf:为何在此处需要加锁？
                    if (isRunning) {
                        waitingEvents.add(packet);
                    } else {
                        processEvent(packet);
                    }
                }
            } else {
                // 就把packet入队。
                waitingEvents.add(packet);
            }
        }

        /**
         * 添加 死亡事件
         */
        public void queueEventOfDeath() {
            waitingEvents.add(eventOfDeath);
        }

        @Override
        public void run() {
            try {
                isRunning = true;
                while (true) {
                    Object event = waitingEvents.take(); // 从队列中获取一个事件
                    if (event == eventOfDeath) { // 如果当前事件为 死亡事件标识
                        wasKilled = true;
                    } else {
                        processEvent(event);
                    }

                    // wasKilled = true后，会继续把队列中等待的任务执行完毕之后，再break跳出循环，从而结束线程执行。
                    if (wasKilled) {
                        synchronized (waitingEvents) {
                            if (waitingEvents.isEmpty()) {
                                isRunning = false;
                                break;
                            }
                        }
                    }
                }
            } catch (InterruptedException e) {
                LOG.error("Event thread exiting due to interruption", e);
            }

            LOG.info("EventThread shut down for session: 0x{}", Long.toHexString(getSessionId()));
        }

        private void processEvent(Object event) {
            try {
                // 如果事件是watcher事件
                if (event instanceof WatcherSetEventPair) {
                    // each watcher will process the event
                    WatcherSetEventPair pair = (WatcherSetEventPair) event;
                    for (Watcher watcher : pair.watchers) {
                        try {
                            watcher.process(pair.event); // 触发watcher
                        } catch (Throwable t) {
                            LOG.error("Error while calling watcher ", t);
                        }
                    }
                } else {
                    // 不是watcher事件的情况，就是处理数据包 packet的情况：（也就是server的response）
                    Packet p = (Packet) event;
                    int rc = 0;
                    String clientPath = p.clientPath;
                    if (p.replyHeader.getErr() != 0) {
                        rc = p.replyHeader.getErr();
                    }
                    // 异步 response 的情况：
                    if (p.cb == null) {
                        // 这里判断响应的callback方法是否存在，如果不存在，是有问题的。
                        LOG.warn("Somehow a null cb got to EventThread!");
                    } else if (p.response instanceof ExistsResponse
                            || p.response instanceof SetDataResponse
                            || p.response instanceof SetACLResponse) {
                        // 获取callback方法
                        StatCallback cb = (StatCallback) p.cb;
                        if (rc == 0) { //
                            if (p.response instanceof ExistsResponse) { // 节点是否存在的响应事件
                                cb.processResult(rc, clientPath, p.ctx, ((ExistsResponse) p.response).getStat());
                            } else if (p.response instanceof SetDataResponse) { // 获取数据value的响应事件
                                cb.processResult(rc, clientPath, p.ctx, ((SetDataResponse) p.response).getStat());
                            } else if (p.response instanceof SetACLResponse) { // 设置acl权限控制的响应事件
                                cb.processResult(rc, clientPath, p.ctx, ((SetACLResponse) p.response).getStat());
                            }
                        } else {
                            // replyCode != 0 的情况：说明是存在异常。
                            cb.processResult(rc, clientPath, p.ctx, null);
                        }
                    } else if (p.response instanceof GetDataResponse) {
                        DataCallback cb = (DataCallback) p.cb;
                        GetDataResponse rsp = (GetDataResponse) p.response;
                        if (rc == 0) {
                            cb.processResult(rc, clientPath, p.ctx, rsp.getData(), rsp.getStat());
                        } else {
                            cb.processResult(rc, clientPath, p.ctx, null,
                                             null);
                        }
                    } else if (p.response instanceof GetACLResponse) {
                        ACLCallback cb = (ACLCallback) p.cb;
                        GetACLResponse rsp = (GetACLResponse) p.response;
                        if (rc == 0) {
                            cb.processResult(rc, clientPath, p.ctx, rsp.getAcl(), rsp.getStat());
                        } else {
                            cb.processResult(rc, clientPath, p.ctx, null,
                                             null);
                        }
                    } else if (p.response instanceof GetChildrenResponse) {
                        ChildrenCallback cb = (ChildrenCallback) p.cb;
                        GetChildrenResponse rsp = (GetChildrenResponse) p.response;
                        if (rc == 0) {
                            cb.processResult(rc, clientPath, p.ctx, rsp.getChildren());
                        } else {
                            cb.processResult(rc, clientPath, p.ctx, null);
                        }
                    } else if (p.response instanceof GetChildren2Response) {
                        Children2Callback cb = (Children2Callback) p.cb;
                        GetChildren2Response rsp = (GetChildren2Response) p.response;
                        if (rc == 0) {
                            cb.processResult(rc, clientPath, p.ctx, rsp.getChildren(), rsp.getStat());
                        } else {
                            cb.processResult(rc, clientPath, p.ctx, null, null);
                        }
                    } else if (p.response instanceof CreateResponse) {
                        StringCallback cb = (StringCallback) p.cb;
                        CreateResponse rsp = (CreateResponse) p.response;
                        if (rc == 0) {
                            cb.processResult(rc, clientPath, p.ctx, (chrootPath == null ? rsp.getPath() : rsp.getPath().substring(chrootPath.length())));
                        } else {
                            cb.processResult(rc, clientPath, p.ctx, null);
                        }
                    } else if (p.response instanceof MultiResponse) {
                        MultiCallback cb = (MultiCallback) p.cb;
                        MultiResponse rsp = (MultiResponse) p.response;
                        if (rc == 0) {
                            List<OpResult> results = rsp.getResultList();
                            int newRc = rc;
                            for (OpResult result : results) {
                                if (result instanceof ErrorResult
                                        && KeeperException.Code.OK.intValue()
                                        != (newRc = ((ErrorResult) result).getErr())) {
                                    break;
                                }
                            }
                            cb.processResult(newRc, clientPath, p.ctx, results);
                        } else {
                            cb.processResult(rc, clientPath, p.ctx, null);
                        }
                    } else if (p.cb instanceof VoidCallback) {
                        VoidCallback cb = (VoidCallback) p.cb;
                        cb.processResult(rc, clientPath, p.ctx);
                    }
                }
            } catch (Throwable t) {
                LOG.error("Caught unexpected throwable", t);
            }
        }
    }

    /**
     *
     * This class services the outgoing request queue and generates the heart
     * beats. It also spawns the ReadThread.
     */
    class SendThread extends ZooKeeperThread {
        /**
         * 最小/最大 ping 读写服务器超时时间
         */
        private final static int minPingRwTimeout = 100;
        private final static int maxPingRwTimeout = 60000;

        private static final int MAX_SEND_PING_INTERVAL = 10000; //10 seconds
        private static final String RETRY_CONN_MSG = ", closing socket connection and attempting reconnect";
        private final ClientCnxnSocket clientCnxnSocket;
        private long lastPingSentNs;
        private Random r = new Random(System.nanoTime());

        // TODO: can not name this method getState since Thread.getState()
        // already exists
        // It would be cleaner to make class SendThread an implementation of
        // Runnable
        private boolean isFirstConnect = true;
        private InetSocketAddress rwServerAddress = null;
        private int pingRwTimeout = minPingRwTimeout;
        // Set to true if and only if constructor of ZooKeeperSaslClient
        // throws a LoginException: see startConnect() below.
        private boolean saslLoginFailed = false;

        SendThread(ClientCnxnSocket clientCnxnSocket) {
            super(makeThreadName("-SendThread()"));
            state = States.CONNECTING;
            this.clientCnxnSocket = clientCnxnSocket;
            setDaemon(true);
        }

        /**
         * 从byteBuffer中读取服务端发来的响应数据。
         *
         * @param incomingBuffer
         * @throws IOException
         */
        void readResponse(ByteBuffer incomingBuffer) throws IOException {
            ByteBufferInputStream bbis = new ByteBufferInputStream(incomingBuffer);
            BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
            ReplyHeader replyHdr = new ReplyHeader();

            replyHdr.deserialize(bbia, "header");
            if (replyHdr.getXid() == -2) {
                // -2 is the xid for pings
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Got ping response for sessionid: 0x" + Long.toHexString(sessionId) + " after " + ((System.nanoTime() - lastPingSentNs) / 1000000) + "ms");
                }
                return;
            }
            if (replyHdr.getXid() == -4) {
                // -4 is the xid for AuthPacket
                if (replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
                    state = States.AUTH_FAILED;
                    eventThread.queueEvent(new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.AuthFailed, null));
                }
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Got auth sessionid:0x" + Long.toHexString(sessionId));
                }
                return;
            }
            if (replyHdr.getXid() == -1) {
                // -1 means notification
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Got notification sessionid:0x" + Long.toHexString(sessionId));
                }
                WatcherEvent event = new WatcherEvent();
                event.deserialize(bbia, "response");

                // 从服务器路径转换为客户端路径   [convert from a server path to a client path]
                if (chrootPath != null) {
                    String serverPath = event.getPath();
                    if (serverPath.compareTo(chrootPath) == 0) {
                        event.setPath("/");
                    } else if (serverPath.length() > chrootPath.length()) {
                        event.setPath(serverPath.substring(chrootPath.length()));
                    } else {
                        LOG.warn("Got server path " + event.getPath() + " which is too short for chroot path " + chrootPath);
                    }
                }

                // 创建通知事件
                WatchedEvent we = new WatchedEvent(event);
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Got " + we + " for sessionid 0x" + Long.toHexString(sessionId));
                }
                // 通知事件 入队列
                eventThread.queueEvent(we);
                return;
            }

            // If SASL authentication is currently in progress, construct and
            // send a response packet immediately, rather than queuing a
            // response as with other packets.
            if (clientTunneledAuthenticationInProgress()) {
                GetSASLRequest request = new GetSASLRequest();
                request.deserialize(bbia, "token");
                zooKeeperSaslClient.respondToServer(request.getToken(), ClientCnxn.this);
                return;
            }

            Packet packet;
            synchronized (pendingQueue) {
                // 如果没有请求正在等待响应，则报错。
                if (pendingQueue.size() == 0) {
                    throw new IOException("Nothing in the queue, but got " + replyHdr.getXid());
                }
                // 取出第一个正在等待的请求包
                packet = pendingQueue.remove();
            }

            // 由于发送请求都是按顺序的，我们最好把当前获得到的response，与第一个request相关联进行处理
            // Since requests are processed in order, we better get a response to the first request!
            try {
                // 如果响应的xid和请求的xid不一样，则报错。
                if (packet.requestHeader.getXid() != replyHdr.getXid()) {
                    packet.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
                    throw new IOException("Xid out of order. Got Xid " + replyHdr.getXid() + " with err " + replyHdr.getErr() + " expected Xid " + packet.requestHeader.getXid() + " for a packet with details: " + packet);
                }

                packet.replyHeader.setXid(replyHdr.getXid());
                packet.replyHeader.setErr(replyHdr.getErr());
                packet.replyHeader.setZxid(replyHdr.getZxid());// 响应的事务id
                if (replyHdr.getZxid() > 0) {
                    // 当前响应的事务id，设置为此客户端最后一个事务id
                    lastZxid = replyHdr.getZxid();
                }

                // 响应数据不为空，并且响应头没有错误，则反序列化响应体
                if (packet.response != null && replyHdr.getErr() == 0) {
                    packet.response.deserialize(bbia, "response");
                }

                if (LOG.isDebugEnabled()) {
                    LOG.debug("Reading reply sessionid:0x" + Long.toHexString(sessionId) + ", packet:: " + packet);
                }
            } finally {
                finishPacket(packet);
            }
        }

        /**
         * Used by ClientCnxnSocket
         *
         * @return
         */
        ZooKeeper.States getZkState() {
            return state;
        }

        ClientCnxnSocket getClientCnxnSocket() {
            return clientCnxnSocket;
        }

        /**
         * <pre>
         * 方法名primeConnection字面意思：首要的连接
         * 方法做的事情：socket连接[ Socket connection established ]建立完毕后，在zk层面首要的需要做的事就是：初始化session。即发送ConnectRequest
         *
         * prime	英[praɪm] 美[praɪm]
         *  adj.	主要的; 首要的; 基本的; 优质的; 上乘的; 优异的; 典型的; 有代表性的;
         *  n.	    盛年; 年富力强的时期; 鼎盛时期; [数] 质数;
         *  v.	    事先指点; 使(某人)做好准备; 把(事物)准备好; 在(金属、木材等上)打底漆;
         *
         * </pre>
         */
        void primeConnection() throws IOException {
            // 从打印的日志可见：socket 链接建立完毕，正在初始化session
            LOG.info("Socket connection established to " + clientCnxnSocket.getRemoteSocketAddress() + ", initiating session");

            isFirstConnect = false;
            // 如果在创建ZooKeeper对象时，指定了sessionId，则使用指定的sessionId；否则，使用0。
            long sessId = (seenRwServerBefore) ? sessionId : 0;

            // 创建 连接请求
            ConnectRequest conReq = new ConnectRequest(0, lastZxid, sessionTimeout, sessId, sessionPasswd);
            synchronized (outgoingQueue) {
                // We add backwards since we are pushing into the front
                // Only send if there's a pending watch
                // TODO: here we have the only remaining use of zooKeeper in
                // this class. It's to be eliminated!
                if (!disableAutoWatchReset) {
                    List<String> dataWatches = zooKeeper.getDataWatches();
                    List<String> existWatches = zooKeeper.getExistWatches();
                    List<String> childWatches = zooKeeper.getChildWatches();
                    if (!dataWatches.isEmpty() || !existWatches.isEmpty() || !childWatches.isEmpty()) {
                        Iterator<String> dataWatchesIter = prependChroot(dataWatches).iterator();
                        Iterator<String> existWatchesIter = prependChroot(existWatches).iterator();
                        Iterator<String> childWatchesIter = prependChroot(childWatches).iterator();
                        long setWatchesLastZxid = lastZxid;

                        while (dataWatchesIter.hasNext() || existWatchesIter.hasNext() || childWatchesIter.hasNext()) {
                            List<String> dataWatchesBatch = new ArrayList<String>();
                            List<String> existWatchesBatch = new ArrayList<String>();
                            List<String> childWatchesBatch = new ArrayList<String>();
                            int batchLength = 0;

                            // Note, we may exceed our max length by a bit when we add the last
                            // watch in the batch. This isn't ideal, but it makes the code simpler.
                            while (batchLength < SET_WATCHES_MAX_LENGTH) {
                                final String watch;
                                if (dataWatchesIter.hasNext()) {
                                    watch = dataWatchesIter.next();
                                    dataWatchesBatch.add(watch);
                                } else if (existWatchesIter.hasNext()) {
                                    watch = existWatchesIter.next();
                                    existWatchesBatch.add(watch);
                                } else if (childWatchesIter.hasNext()) {
                                    watch = childWatchesIter.next();
                                    childWatchesBatch.add(watch);
                                } else {
                                    break;
                                }
                                batchLength += watch.length();
                            }

                            // 设置watcher请求。
                            SetWatches sw = new SetWatches(setWatchesLastZxid, dataWatchesBatch, existWatchesBatch, childWatchesBatch);
                            RequestHeader h = new RequestHeader();
                            h.setType(ZooDefs.OpCode.setWatches);
                            h.setXid(-8); // xid = -8 表示：setWatches请求
                            Packet packet = new Packet(h, new ReplyHeader(), sw, null, null);

                            outgoingQueue.addFirst(packet); // ZooDefs.OpCode.setWatches 请求放到第一位
                        }
                    }
                }

                for (AuthData id : authInfo) {
                    Packet authPacket = new Packet(new RequestHeader(-4, OpCode.auth), null, new AuthPacket(0, id.scheme, id.data), null, null);
                    outgoingQueue.addFirst(authPacket); // authInfo请求放到第一位
                }
                Packet createSessionPacket = new Packet(null, null, conReq, null, null, readOnly);
                outgoingQueue.addFirst(createSessionPacket); // 创建session请求，放到第一位。【注意：只有创建session请求的数据包，requestHeader为null】
            }

            // 创建session请求/创建session 完毕，从此在socket层面只关心read和write操作
            clientCnxnSocket.enableReadWriteOnly();

            if (LOG.isDebugEnabled()) {
                LOG.debug("Session establishment request sent on " + clientCnxnSocket.getRemoteSocketAddress());
            }
        }

        private List<String> prependChroot(List<String> paths) {
            if (chrootPath != null && !paths.isEmpty()) {
                for (int i = 0; i < paths.size(); ++i) {
                    String clientPath = paths.get(i);
                    String serverPath;
                    // handle clientPath = "/"
                    if (clientPath.length() == 1) {
                        serverPath = chrootPath;
                    } else {
                        serverPath = chrootPath + clientPath;
                    }
                    paths.set(i, serverPath);
                }
            }
            return paths;
        }

        private void sendPing() {
            lastPingSentNs = System.nanoTime();
            RequestHeader h = new RequestHeader(-2, OpCode.ping);
            queuePacket(h, null, null, null, null, null, null, null, null);
        }

        private void startConnect() throws IOException {
            // initializing it for new connection
            saslLoginFailed = false;
            // 正在链接状态。
            state = States.CONNECTING;

            InetSocketAddress addr;
            if (rwServerAddress != null) { // 如果‘可读写服务端地址’不为空，则就使用此服务端地址。
                addr = rwServerAddress;
                rwServerAddress = null;
            } else {
                addr = hostProvider.next(1000); // 否则，获取下一个server连接地址
            }

            // 设置线程名称。
            setName(getName().replaceAll("\\(.*\\)","(" + addr.getHostName() + ":" + addr.getPort() + ")"));
            if (ZooKeeperSaslClient.isEnabled()) {
                try {
                    String principalUserName = System.getProperty(ZK_SASL_CLIENT_USERNAME, "zookeeper");
                    zooKeeperSaslClient = new ZooKeeperSaslClient(principalUserName + "/" + addr.getHostName());
                } catch (LoginException e) {
                    // An authentication error occurred when the SASL client tried to initialize:
                    // for Kerberos this means that the client failed to authenticate with the KDC.
                    // This is different from an authentication error that occurs during communication
                    // with the Zookeeper server, which is handled below.
                    LOG.warn("SASL configuration failed: " + e + " Will continue connection to Zookeeper server without " + "SASL authentication, if Zookeeper server allows it.");
                    eventThread.queueEvent(new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.AuthFailed, null));
                    saslLoginFailed = true;
                }
            }
            logStartConnect(addr);

            clientCnxnSocket.connect(addr);
        }

        private void logStartConnect(InetSocketAddress addr) {
            String msg = "Opening socket connection to server " + addr;
            if (zooKeeperSaslClient != null) {
                msg += ". " + zooKeeperSaslClient.getConfigStatus();
            }
            LOG.info(msg);
        }
        @Override
        public void run() {
            // 线程启动后，立即把相关数据保存到 clientCnxnSocket 中。比如：sessionId、now（即启动事件）
            clientCnxnSocket.introduce(this, sessionId);
            clientCnxnSocket.updateNow();
            clientCnxnSocket.updateLastSendAndHeard();
            int timeout;
            long lastPingRwServer = Time.currentElapsedTime();
            while (state.isAlive()) {
                try {
                    // 【1、】如果：socket尚未连接：
                    if (!clientCnxnSocket.isConnected()) {
                        if (!isFirstConnect) {
                            try {
                                Thread.sleep(r.nextInt(1000));
                            } catch (InterruptedException e) {
                                LOG.warn("Unexpected exception", e);
                            }
                        }
                        // 不再重新建立链接，如果已经设置为closing状态
                        if (closing || !state.isAlive()) {
                            break;
                        }
                        // 开始链接server
                        startConnect();
                        clientCnxnSocket.updateLastSendAndHeard();
                    }


                    // 【2、】如果是‘已连接状态’
                    if (state.isConnected()) {
                        // 如果 zooKeeperSaslClient 不为null，则去 确定是否需要发送Auth失败事件。
                        if (zooKeeperSaslClient != null) {
                            boolean sendAuthEvent = false;
                            if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) {
                                try {
                                    zooKeeperSaslClient.initialize(ClientCnxn.this);
                                } catch (SaslException e) {
                                    LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e);
                                    state = States.AUTH_FAILED;
                                    sendAuthEvent = true;
                                }
                            }

                            KeeperState authState = zooKeeperSaslClient.getKeeperState();
                            if (authState != null) {
                                if (authState == KeeperState.AuthFailed) {
                                    // An authentication error occurred during authentication with the Zookeeper Server.
                                    state = States.AUTH_FAILED;
                                    sendAuthEvent = true;
                                } else {
                                    if (authState == KeeperState.SaslAuthenticated) {
                                        sendAuthEvent = true;
                                    }
                                }
                            }
                            // 发送权限事件
                            if (sendAuthEvent) {
                                eventThread.queueEvent(new WatchedEvent(Watcher.Event.EventType.None, authState, null));
                            }
                        }

                        // read超时 事件
                        timeout = readTimeout - clientCnxnSocket.getIdleRecv();
                    } else {
                        // 【3、】‘尚未连接’状态。           else表示：   !state.isConnected()
                        // 连接超时事件
                        timeout = connectTimeout - clientCnxnSocket.getIdleRecv();
                    }
                    // 如果read超时 或者 connect超时，则报错
                    if (timeout <= 0) {
                        String warnInfo = "Client session timed out, have not heard from server in " + clientCnxnSocket.getIdleRecv() + "ms " +
                                "for sessionid 0x" + Long.toHexString(sessionId);
                        LOG.warn(warnInfo);
                        throw new SessionTimeoutException(warnInfo);
                    }
                    // ==============================>  未超时：


                    // 【4、】如果已经连接：判断是否需要发送ping请求
                    if (state.isConnected()) {
                        //1000(1 second) is to prevent race condition missing to send the second ping
                        // 译：1秒是为了防止‘静态条件消失race condition missing’而发送第二个ping      todo-lxf:静态条件？？？这里是在考虑什么情况的？
                        //also make sure not to send too many pings when readTimeout is small
                        // 译：也是为了确保不要发送太多的ping请求，当readTimeout设置太小时
                        int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend() -
                                ((clientCnxnSocket.getIdleSend() > 1000) ? 1000 : 0);
                        //send a ping request either time is due or no packet sent out within MAX_SEND_PING_INTERVAL
                        if (timeToNextPing <= 0 || clientCnxnSocket.getIdleSend() > MAX_SEND_PING_INTERVAL) {
                            // ping服务端
                            sendPing();
                            // 修改最后发送时间
                            clientCnxnSocket.updateLastSend();
                        } else {
                            if (timeToNextPing < timeout) {
                                timeout = timeToNextPing;
                            }
                        }
                    }

                    // 【5、】如果当前连接的是read-only服务端，则
                    // 去寻找 read/write server，并把read/write server的域名，保存到 rwServerAddress变量，以便下一次试图连接时使用.详见：pingRwServer();
                    if (state == States.CONNECTEDREADONLY) {
                        long now = Time.currentElapsedTime();
                        int idlePingRwServer = (int) (now - lastPingRwServer);
                        if (idlePingRwServer >= pingRwTimeout) {
                            lastPingRwServer = now;
                            idlePingRwServer = 0;
                            pingRwTimeout = Math.min(2 * pingRwTimeout, maxPingRwTimeout);
                            pingRwServer();
                        }
                        timeout = Math.min(timeout, pingRwTimeout - idlePingRwServer);
                    }

                    // ==============================> 处理事务 <==================================================
                    clientCnxnSocket.doTransport(timeout, pendingQueue, outgoingQueue, ClientCnxn.this); // 【处理事务！！！】
                    // ==============================> 处理事务 <==================================================
                } catch (Throwable e) {
                    if (closing) {
                        if (LOG.isDebugEnabled()) {
                            // closing so this is expected
                            LOG.debug("An exception was thrown while closing send thread for session 0x" + Long.toHexString(getSessionId()) + " : " + e.getMessage());
                        }
                        break;
                    } else {
                        // this is ugly, you have a better way speak up
                        if (e instanceof SessionExpiredException) {
                            LOG.info(e.getMessage() + ", closing socket connection");
                        } else if (e instanceof SessionTimeoutException) {
                            LOG.info(e.getMessage() + RETRY_CONN_MSG);
                        } else if (e instanceof EndOfStreamException) {
                            LOG.info(e.getMessage() + RETRY_CONN_MSG);
                        } else if (e instanceof RWServerFoundException) {
                            LOG.info(e.getMessage());
                        } else {
                            LOG.warn("Session 0x" + Long.toHexString(getSessionId()) + " for server " + clientCnxnSocket.getRemoteSocketAddress() + ", unexpected error" + RETRY_CONN_MSG, e);
                        }
                        // 出现异常后：清理
                        cleanup();
                        // 出现异常后：如果当前状态是alive，则发送 Disconnected 事件
                        if (state.isAlive()) {
                            eventThread.queueEvent(new WatchedEvent(Event.EventType.None, Event.KeeperState.Disconnected, null));
                        }

                        // 更新时间
                        clientCnxnSocket.updateNow();
                        clientCnxnSocket.updateLastSendAndHeard();
                    }
                }
            } // -- while(state.isAlive()) -- end --


            // 跳出while循环：清理、关闭socket
            cleanup();
            clientCnxnSocket.close();

            // 如果此时客户端还活着，则发送 Disconnected 事件。（一般来说客户端收到此事件之后，都会在重新new一个ZooKeeper客户端。）
            if (state.isAlive()) {
                eventThread.queueEvent(new WatchedEvent(Event.EventType.None, Event.KeeperState.Disconnected, null));
            }
            ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(),"SendThread exited loop for session: 0x" + Long.toHexString(getSessionId()));
        }

        private void pingRwServer() throws RWServerFoundException {
            String result = null;
            InetSocketAddress addr = hostProvider.next(0);
            LOG.info("Checking server " + addr + " for being r/w. Timeout " + pingRwTimeout);

            Socket sock = null;
            BufferedReader br = null;
            try {
                sock = new Socket(addr.getHostName(), addr.getPort());
                sock.setSoLinger(false, -1);
                sock.setSoTimeout(1000);
                sock.setTcpNoDelay(true);
                sock.getOutputStream().write("isro".getBytes()); // socket链接上服务端，发送isro四字命令
                sock.getOutputStream().flush();
                sock.shutdownOutput();
                br = new BufferedReader(new InputStreamReader(sock.getInputStream()));
                result = br.readLine(); // 获得结果
            } catch (ConnectException e) {
                // ignore, this just means server is not up
            } catch (IOException e) {
                // some unexpected error, warn about it
                LOG.warn("Exception while seeking for r/w server " + e.getMessage(), e);
            } finally {
                if (sock != null) {
                    try {
                        sock.close();
                    } catch (IOException e) {
                        LOG.warn("Unexpected exception", e);
                    }
                }
                if (br != null) {
                    try {
                        br.close();
                    } catch (IOException e) {
                        LOG.warn("Unexpected exception", e);
                    }
                }
            }

            // 如果isro四字命令返回结果为rw，则说明当前ping的server就是可读可写服务端
            if ("rw".equals(result)) {
                pingRwTimeout = minPingRwTimeout;
                // 保存找到的‘可读写server’地址，以便下一次试图连接时使用
                rwServerAddress = addr;
                throw new RWServerFoundException("Majority server found at " + addr.getHostName() + ":" + addr.getPort());
            }
        }

        private void cleanup() {
            clientCnxnSocket.cleanup();
            synchronized (pendingQueue) {
                for (Packet p : pendingQueue) {
                    conLossPacket(p);
                }
                pendingQueue.clear();
            }
            synchronized (outgoingQueue) {
                for (Packet p : outgoingQueue) {
                    conLossPacket(p);
                }
                outgoingQueue.clear();
            }
        }

        /**
         * Callback invoked by the ClientCnxnSocket once a connection has been
         * established.
         *
         * @param _negotiatedSessionTimeout
         * @param _sessionId
         * @param _sessionPasswd
         * @param isRO
         * @throws IOException
         */
        void onConnected(int _negotiatedSessionTimeout, long _sessionId,byte[] _sessionPasswd, boolean isRO) throws IOException {
            negotiatedSessionTimeout = _negotiatedSessionTimeout;
            if (negotiatedSessionTimeout <= 0) {
                state = States.CLOSED;

                eventThread.queueEvent(new WatchedEvent(Watcher.Event.EventType.None, Watcher.Event.KeeperState.Expired, null));
                eventThread.queueEventOfDeath();

                String warnInfo;
                warnInfo = "Unable to reconnect to ZooKeeper service, session 0x" + Long.toHexString(sessionId) + " has expired";
                LOG.warn(warnInfo);
                throw new SessionExpiredException(warnInfo);
            }
            if (!readOnly && isRO) {
                LOG.error("Read/write client got connected to read-only server");
            }
            readTimeout = negotiatedSessionTimeout * 2 / 3;
            connectTimeout = negotiatedSessionTimeout / hostProvider.size();
            hostProvider.onConnected();
            sessionId = _sessionId;
            sessionPasswd = _sessionPasswd;
            // 根据isRO判断已链接状态
            state = (isRO) ? States.CONNECTEDREADONLY : States.CONNECTED;
            seenRwServerBefore |= !isRO;
            LOG.info("Session establishment complete on server " + clientCnxnSocket.getRemoteSocketAddress() + ", sessionid = 0x" + Long.toHexString(sessionId) + ", negotiated timeout = " + negotiatedSessionTimeout + (isRO ? " (READ-ONLY mode)" : ""));

            KeeperState eventState = (isRO) ? KeeperState.ConnectedReadOnly : KeeperState.SyncConnected;
            eventThread.queueEvent(new WatchedEvent(Watcher.Event.EventType.None, eventState, null));
        }

        void close() {
            state = States.CLOSED;
            clientCnxnSocket.wakeupCnxn();
        }

        void testableCloseSocket() throws IOException {
            clientCnxnSocket.testableCloseSocket();
        }

        public boolean clientTunneledAuthenticationInProgress() {
            // 1. SASL client is disabled.
            if (!ZooKeeperSaslClient.isEnabled()) {
                return false;
            }

            // 2. SASL login failed.
            if (saslLoginFailed == true) {
                return false;
            }

            // 3. SendThread has not created the authenticating object yet,
            // therefore authentication is (at the earliest stage of being) in progress.
            if (zooKeeperSaslClient == null) {
                return true;
            }

            // 4. authenticating object exists, so ask it for its progress.
            return zooKeeperSaslClient.clientTunneledAuthenticationInProgress();
        }

        public void sendPacket(Packet p) throws IOException {
            clientCnxnSocket.sendPacket(p);
        }
    }
}
